package utils

import (
	"encoding/json"
	"errors"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/robfig/cron"
	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/threading"
	"mzpd_server/internal/svc"
	"strings"
	"sync"
)

type MqttServer struct {
	client    mqtt.Client
	serverCtx *svc.ServiceContext
	IsRun     bool
	Cron      *cron.Cron
}

var instance *MqttServer
var once sync.Once

func GetMqttServer() *MqttServer {
	once.Do(func() {
		instance = &MqttServer{
			IsRun: false,
			Cron:  cron.New(),
		}
	})
	return instance
}

// messagePubHandler 接收消息
func (mServer *MqttServer) messagePubHandler(client mqtt.Client, msg mqtt.Message) {
	topic := msg.Topic()
	payload := msg.Payload()
	fmt.Printf("Received message: %s from topic: %s\n", payload, topic)
	//这里收到消息后,操作 rpc 获取/处理 数据,mServer.serverCtx 可以拿到对应的 rpc 相关配置信息 TODO...
	//接收科室/诊室/叫号/接诊相关信息处理 TODO...
	if strings.Contains(topic, "$SYS/brokers") { //是上下线事件
		var status OnlineStatus
		json.Unmarshal(payload, &status)
		fmt.Printf("状态：%v", &status)
		threading.GoSafe(func() { //发送http更新设备状态
			//var statusUp = UpdateStatus{
			//	DeviceId: status.Clientid,
			//	Status:   0,
			//}
			//if status.CleanStart {
			//	statusUp.Status = 0
			//} else {
			//	statusUp.Status = 1
			//}
			//marshal, _ := json.Marshal(statusUp)
			//var httpclient = Httpclient{}
			//httpclient.Post("http://127.0.0.1:8889/v/device/updateStatus", string(marshal))
		})
	}
}

// connectHandler 连接成功
func (mServer *MqttServer) connectHandler(client mqtt.Client) {
	fmt.Println("Connected")
	mServer.IsRun = true
}

// connectLostHandler 断开
func (mServer *MqttServer) connectLostHandler(client mqtt.Client, err error) {
	fmt.Printf("Connect lost: %v", err)
	mServer.IsRun = false
}

// PublishHandler 发布消息
func (mServer *MqttServer) PublishHandler(topic string, qos byte, retained bool, payload string) {
	if !mServer.IsRun { //未启动
		//_, err := mServer.Start(mServer.serverCtx)
		//if err != nil {
		//	logx.Errorf("MQTT 启动失败%s", err.Error())
		//	return
		//}
		//if !mServer.IsRun { //未启动
		//	logx.Errorf("MQTT 启动失败%s", err.Error())
		//	return
		//}
		logx.Errorf("MQTT 未启动%s", mServer.IsRun)
		return
	}
	token := mServer.client.Publish(topic, qos, retained, payload)
	wait := token.Wait()
	if wait {
	}
	logx.Infof("[Publish]:%v", payload)
}

// SubscribeMultiple 订阅多个消息
func (mServer *MqttServer) SubscribeMultiple(topic string, qos byte, callback mqtt.MessageHandler) {
	mServer.client.(mqtt.Client).Subscribe(topic, qos, callback)
}

// subscribe 订阅单个消息
func (mServer *MqttServer) subscribe(client mqtt.Client, err error) {
	fmt.Printf("Connect lost: %v", err)
}

// Start 开启服务
func (mServer *MqttServer) Start(serverCtx *svc.ServiceContext) (bool, error) {
	if mServer.IsRun {
		return false, errors.New("已经启动了")
	}
	mServer.serverCtx = serverCtx
	config := serverCtx.Config

	var broker = config.Broker   // "127.0.0.1"
	var port = config.BrokerPort // 1883
	opts := mqtt.NewClientOptions()
	opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))
	opts.SetClientID(config.ClientID)
	opts.SetUsername(config.UserName)
	opts.SetPassword("123456")
	opts.SetDefaultPublishHandler(mServer.messagePubHandler)
	opts.OnConnect = mServer.connectHandler
	opts.OnConnectionLost = mServer.connectLostHandler
	opts.AutoReconnect = true
	client := mqtt.NewClient(opts)
	mServer.client = client
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		err1 := token.Error()
		fmt.Printf("MQTT Starting server at Fail:%s", err1.Error())
		logx.Errorf("MQTT Starting server at Fail:%s", err1.Error())

		mServer.serverCtx.Redis.Set("MQ_STATUS", "0")
		return false, token.Error()
	}
	topic := "testtopic/#"
	filters := map[string]byte{}
	filters[topic] = byte(2)
	//设备上下线通知
	filters["$SYS/brokers/+/clients/+/+"] = byte(2)
	token := client.SubscribeMultiple(filters, nil)
	token.Wait()
	mServer.IsRun = true
	fmt.Printf("MQTT Starting server Success")
	mServer.serverCtx.Redis.Set("MQ_STATUS", "1")
	return true, nil
}
